package com.study.spark.utils;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;

import com.alibaba.fastjson.JSONObject;
import com.study.spark.conf.ConfigurationManager;
import com.study.spark.constant.Constants;
import com.study.spark.data.MockData;

/**
 * Spark工具类
 * 
 * @author Administrator
 *
 */
public class SparkUtil {

  /**
   * 根据当前是否本地测试的配置 决定，如何设置SparkConf的master
   */
  public static void setMaster(SparkConf conf) {
    boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
    if (local) {
      conf.setMaster("local");
    }
  }

  /**
   * 获取SQLContext 如果spark.local设置为true，那么就创建SQLContext；否则，创建HiveContext
   * 
   * @param sc
   * @return
   */
  public static SQLContext getSQLContext(SparkContext sc) {
    return ConfigurationManager.getBoolean(Constants.SPARK_LOCAL) ? new SQLContext(sc)
        : new HiveContext(sc);
  }

  /**
   * 生成模拟数据 如果spark.local配置设置为true，则生成模拟数据；否则不生成
   * 
   * @param sc
   * @param sqlContext
   */
  public static void mockData(JavaSparkContext sc, SQLContext sqlContext) {
    boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
    if (local) {
      MockData.mock(sc, sqlContext);
    }
  }

  /**
   * 获取指定日期范围内的用户行为数据RDD
   * 
   * @param sqlContext
   * @param taskParam
   * @return
   */
  public static JavaRDD<Row> getActionRDDByDateRange(SQLContext sqlContext, JSONObject taskParam) {
    String startDate = ParamUtil.getParam(taskParam, Constants.PARAM_START_DATE);
    String endDate = ParamUtil.getParam(taskParam, Constants.PARAM_END_DATE);

    String sql = "select * " + "from user_visit_action " + "where date>='" + startDate + "' "
        + "and date<='" + endDate + "'";

    DataFrame actionDF = sqlContext.sql(sql);

    return actionDF.javaRDD();
  }

}
